#include "umSink.h"
#include "httpTools.h"
#include <signal.h>
#include "umlog.h"
#include "lstack.h"
#include "hashmap.h"
#include "umapi_libc.h"
#include "umapi_file.h"

typedef struct
{
	UM_HANDLE fd;
	UMAPI_SOCKET_ADDR ipaddr;
	struct umapi_list_head head;
}CientNode;

typedef struct
{
	UM_CHAR name[100];
	UM_HANDLE tmpFile;
	UM_S32 len;
	UM_S64 tick;
	UM_S32 duration;
}S_TS_BLOCK;


typedef struct
{
	UM_HANDLE fd;
	UM_HANDLE mutexHandle;
	UM_EFD *efd;
	UMAPI_SOCKET_ADDR addr;
	UM_VOID *transcode;
	UM_VOID *s;
	UM_CHAR dest[200];
	UM_S32 cacheBuffIndex;
	UM_U8 cacheBuff[MAXTCCACHESIZE];
	struct umapi_list_head head;
}S_HLSNODE;
#define MAX_TS_BLOCK_COUNT 6
//#define USEREMOTEKEYSERVER 1
typedef struct 
{
	UM_CHAR dest[100]; 
	UM_HANDLE mutexHandle;
	UM_VOID *s;
	UM_BOOL enckey;
	UM_U8 key[20];
	UM_U8 iv[20];
	UM_S32 keytype;
	UM_S32 sequence;
	UM_CHAR m3u8[4096];
	S_TS_BLOCK *block[MAX_TS_BLOCK_COUNT];
	UM_S32 blockIndex;
	UM_S32 blockId;
	S_TS_BLOCK *curBlock;
	struct umapi_list_head ch;		 
	struct umapi_list_head head;	
}Trancode;

typedef struct
{
	pid_t pid;
	UM_S16 port;
	UM_S16 nn;
	UM_S32 localport;
	UM_CHAR host[20];	
	UM_CHAR path[200];
	UM_CHAR m3u8path[100];
	UM_VOID *epoll;
	S_SINK_WORK workCall;
	UM_HANDLE fd;
	UM_HANDLE taskHandle,mutexHandle;
#if USE_HASH
	map_t *map;
#endif	
	UM_S32 runflag, stopflag;
	struct umapi_list_head head;
	struct umapi_list_head clients;
	struct umapi_list_head paths;
}S_HLSSERVER;

typedef struct
{	
	UM_S16 port;
	UM_CHAR host[20];	
	UM_CHAR path[200];
	UM_VOID *n;
	S_HLSSERVER *server;
	struct umapi_list_head head;
	
}S_HLSSINK;

#define TARGET "HLSSINK"
#define DEBUGLEVEL hlsLevel
static int hlsLevel =  0;
static UM_U8 *firstTsBuff = UM_NULL;
static UM_S32 firstTsBuffSize;

void setHlsDebug(int level)
{
	hlsLevel = level;
}


INITSERVER(servers,serverMutexHandle, initflag) ;

#define M3U8 "application/x-mpegURL"
#define TS "video/MP2T"
#define HLSSERVERNAME "umHlsServer"

static UM_VOID writeKey(UM_U8 *key, UM_CHAR *name)
{
	UM_HANDLE tmpFile;
	UM_CHAR tmpFilename[100];
	umapi_sprintf(tmpFilename, "/tmp/ts/%s/hls.key", name);
	if(UM_SUCCESS != umapi_file_fopen(tmpFilename, "wb", &tmpFile))
	{
		perror(name);
		return ;
	}
	umapi_file_fwrite(key, 1, 16, tmpFile);
	umapi_file_fclose(tmpFile);
	return ;
}
static UM_VOID writeM3u8File(UM_CHAR *m3u8, UM_CHAR *name)
{
	UM_HANDLE tmpFile;
	UM_CHAR tmpFilename[100];
	umapi_sprintf(tmpFilename, "/tmp/ts/%s/index.m3u8", name);
	if(UM_SUCCESS != umapi_file_fopen(tmpFilename, "wb", &tmpFile))
	{
		perror(name);
		return ;
	}
	umapi_file_fwrite(m3u8, 1, umapi_strlen(m3u8), tmpFile);
	umapi_file_fclose(tmpFile);
	return ;
}


static Trancode* findTrancode(UM_VOID *s, UM_CHAR *url)
{
	Trancode* ret = UM_NULL;
	S_HLSSERVER * c = (S_HLSSERVER *)s;
	umapi_os_mutex_lock(c->mutexHandle, -1);
#if USE_HASH
	hashmap_get(c->map, url, &ret);
#else
	struct umapi_list_head *pos, *n;
	umapi_list_for_each_safe(pos, n, &c->paths)
	{
		Trancode* t =  umapi_list_entry(pos, Trancode, head);
		if(umapi_strncmp(url, t->dest, umapi_strlen(t->dest)) == 0
			|| umapi_strncmp(&url[1], t->dest, umapi_strlen(t->dest)) == 0)
		{
			ret = t;
			break;
		}
	}
#endif	
	umapi_os_mutex_unlock(c->mutexHandle);
	return ret;
}
static UM_U8 readU8(UM_CHAR *pData)
{
	UM_CHAR kk[3] = {0};
	kk[0] = pData[0];
	kk[1] = pData[1];
	kk[2] = '\0';
	return  strtol(kk, NULL, 16);
}
static UM_CHAR * getPlayId(UM_CHAR *src)
{
	UM_CHAR *p = src, *p1;
	while(1)
	{
		p1 = umapi_strstr(p, "/");
		if(!p1)
			return p ;
		else
			p = p1 + 1;
	}
	return UM_NULL;
}
static UM_VOID updateBlocks(Trancode* ptnc);
static UM_VOID initBlock(S_TS_BLOCK *pBlock, Trancode* ptnc);

#define FIRSTTSBUFF "./firstTs.ts"
#define MAX_FIRST_TS_COUNT 1
static UM_VOID addFirstBlock(Trancode* pNode)
{
	for(int i=0;i<MAX_FIRST_TS_COUNT;i++)
	{
		pNode->curBlock = pNode->block[pNode->blockIndex++];
		initBlock(pNode->curBlock, pNode);
		if(pNode->curBlock->tmpFile)
			umapi_file_fwrite(firstTsBuff, 1, firstTsBuffSize, pNode->curBlock->tmpFile);
		pNode->curBlock->duration = 1000;
		updateBlocks(pNode);
	}
	return ;
}
static UM_VOID initFirstTsBuff()
{
	UM_HANDLE fileHandle;
	umapi_file_fopen(FIRSTTSBUFF, "rb", &fileHandle);
	umapi_file_fseek(fileHandle, 0, UM_FILE_SEEK_END);
	firstTsBuffSize = umapi_file_ftell(fileHandle);
	firstTsBuff = (UM_U8*)umapi_malloc(firstTsBuffSize);
	umapi_file_fseek(fileHandle, 0, UM_FILE_SEEK_SET);
	umapi_file_fread(firstTsBuff, 1, firstTsBuffSize, fileHandle);
	umapi_file_close(fileHandle);
	return;
}
static  UM_VOID* addClient(UM_VOID *s, UM_CHAR *url)
{
	S_HLSSERVER *c = (S_HLSSERVER*)s;
	Trancode* t = findTrancode(s, url);
	if(t)
	{
		return t;
	}
	t = (Trancode*)umapi_malloc(sizeof(Trancode));
	if(!t)
	{
		LOGS("malloc Trancode fail");
		return UM_FAILURE;
	}
	umapi_memset(t, 0, sizeof(Trancode));
	t->s = s;
	UM_CHAR *pkey = UM_NULL;
	if(pkey = umapi_strstr(url, "hlskey="))
	{
		t->enckey =1;
		pkey += 7;
		for(int i = 0; i<16 ;i++)
		{
			t->key[i] = readU8(pkey);
			pkey +=2 ;
		}
	}
	else
		t->enckey = 0;
	umapi_snprintf(t->dest, sizeof(t->dest), "%s", url);
	pkey = umapi_strstr(url, "?");
	if(pkey)
	   t->dest[pkey - url] = '\0';
	UM_CHAR tmpDir[1024];
	umapi_sprintf(tmpDir, "mkdir /tmp/ts/%s -p", t->dest);
	LOGD("create new Node :%s hlskey:%d url:%s", tmpDir, t->enckey,url);
	//umapi_dir_create(tmpDir);
	system(tmpDir);
	//runcmd("sync");
	umapi_os_mutex_create(UM_NULL,UM_OS_MUTEX_ATTR_TIMED_NP,&t->mutexHandle);
	umapi_os_mutex_lock(c->mutexHandle, -1);
	umapi_list_create(&t->ch);
	for(int i=0;i <MAX_TS_BLOCK_COUNT; i++)
	{
		t->block[i] = (S_TS_BLOCK*)umapi_malloc(sizeof(S_TS_BLOCK));
		umapi_memset(t->block[i], 0, sizeof(S_TS_BLOCK));
	}
	t->blockIndex = 0 ;
	t->curBlock = UM_NULL;
	t->keytype = 128;
	//createRandom(t->key, t->keytype / 8);
	createRandom(t->iv, t->keytype / 8);
	
#if USE_HASH
	hashmap_put(c->map, t->dest, t);
	LOGD("add hls node :%s", t->dest);
#else
	umapi_list_add(&t->head, &c->paths);
#endif		
	UM_CHAR m3u8Buff[4096];
	UM_CHAR m3u8KeyLine[80];
	if(t->enckey)
		umapi_sprintf(m3u8KeyLine, "#EXT-X-KEY:METHOD=AES-%d,URI=\"http://%s:%d%s%s\"\r\n",
			t->keytype,
			gSystemData.hlsHost,
			gSystemData.hlsPort,
			gSystemData.hlshandlePath,
			getPlayId(t->dest));
	else
		m3u8KeyLine[0] = '\0';
	umapi_sprintf(m3u8Buff, "#EXTM3U\r\n"
			"#EXT-X-MEDIA-SEQUENCE:0\r\n"
			"#EXT-X-ALLOW-CACHE:NO\r\n"
			"#EXT-X-VERSION:0\r\n"
			"%s"
			"#EXT-X-TARGETDURATION:1\r\n",
			m3u8KeyLine);
	umapi_sprintf(t->m3u8, "%s", m3u8Buff);
	LOGI("add Trancode :%p ,url:%s,mutexHandle:%p, m3u8Buff:%s", t, url,t->mutexHandle, m3u8Buff);
	umapi_os_mutex_unlock(c->mutexHandle);

	addFirstBlock(t);
	writeKey(t->key, t->dest);
	return t;
}
static UM_S32 delClient(UM_VOID *s, UM_VOID* n)
{
	Trancode* ptnc = n;
	if(ptnc)
	{
		S_HLSSERVER *hs = (S_HLSSERVER *)ptnc->s;
		umapi_os_mutex_lock(ptnc->mutexHandle,-1);
		struct umapi_list_head *pos, *n;
		umapi_list_for_each_safe(pos, n, &ptnc->ch)
		{
			S_HLSNODE *tc = umapi_list_entry(pos,S_HLSNODE, head);
			umapi_list_del(&tc->head);
			umapi_socket_close(tc->fd);
			tc->transcode = UM_NULL;			
		}
		umapi_os_mutex_lock(hs->mutexHandle, -1);
#if USE_HASH
		hashmap_remove(hs->map,ptnc->dest);
		LOGD("remove hls node :%s", ptnc->dest);
#else
		umapi_list_del(&ptnc->head);
#endif
		umapi_os_mutex_unlock(hs->mutexHandle);
		umapi_os_mutex_unlock(ptnc->mutexHandle);
		umapi_os_mutex_destroy(ptnc->mutexHandle);
#if 0		
		UM_CHAR tmpFilename[120];
		for(int i =0; i<MAX_TS_BLOCK_COUNT; i++)
		{
			S_TS_BLOCK * pBlock = ptnc->block[i];
			if(umapi_strlen(pBlock->name) > 0)
			{
				umapi_sprintf(tmpFilename, "/tmp/ts/%s.ts",pBlock->name);
				umapi_file_remove(tmpFilename);
				if(ptnc->enckey){
					umapi_sprintf(tmpFilename, "/tmp/ts/%s",pBlock->name);
					umapi_file_remove(tmpFilename);
				}
			}
		}
		umapi_sprintf(tmpFilename, "/tmp/ts/%s/index.m3u8", ptnc->dest);
		umapi_file_remove(tmpFilename);
		umapi_sprintf(tmpFilename, "/tmp/ts/%s/hls.key", ptnc->dest);
		umapi_file_remove(tmpFilename);
		
#else
		UM_CHAR tmpDir[1024];
		umapi_sprintf(tmpDir, "/tmp/ts/%s", ptnc->dest);
		umapi_dir_remove(tmpDir, UM_FALSE);
		for(int i =0; i<MAX_TS_BLOCK_COUNT; i++)
		{
			S_TS_BLOCK * pBlock = ptnc->block[i];
			if(pBlock->tmpFile)
				umapi_file_fclose(pBlock->tmpFile);
		}
#endif
		umapi_free(ptnc);
	}
	return UM_SUCCESS;

}
static UM_S32 startClient(UM_VOID *s, UM_VOID* n)
{
	return UM_SUCCESS;
}
static UM_S32 stopClient(UM_VOID *s, UM_VOID* n)
{
	return UM_SUCCESS;
}

static UM_VOID initBlock(S_TS_BLOCK *pBlock, Trancode* ptnc)
{
	UM_CHAR tmpFilename[100];
	if(umapi_strlen(pBlock->name) > 0)
	{
		umapi_sprintf(tmpFilename, "/tmp/ts/%s.ts",pBlock->name);
		umapi_file_remove(tmpFilename);
		if(ptnc->enckey){
			umapi_sprintf(tmpFilename, "/tmp/ts/%s",pBlock->name);
			umapi_file_remove(tmpFilename);
			pBlock->tmpFile = UM_NULL;
		}
	}
	umapi_sprintf(pBlock->name, "%s/%d", ptnc->dest, ptnc->blockId++);
	if(!ptnc->enckey)
		umapi_sprintf(tmpFilename, "/tmp/ts/%s.ts", pBlock->name);
	else
		umapi_sprintf(tmpFilename, "/tmp/ts/%s", pBlock->name);
	umapi_file_fopen(tmpFilename, "wb", &pBlock->tmpFile);
	pBlock->duration = 0;
	pBlock->len = 0;
	pBlock->tick = umapi_os_get_tick();
}
static UM_VOID updateM3u8(S_TS_BLOCK* item, UM_VOID *param)
{
	if(item->duration == 0)
		return;
	UM_CHAR *pBuff = (UM_CHAR*)param;
	UM_F f = (UM_F)(item->duration) / 1000;
	umapi_sprintf(&pBuff[umapi_strlen(pBuff)], "#EXTINF:%.2f,\r\nhttp://%s:%d/ts%s.ts\r\n",
		f, 
		gSystemData.appPulicHost,
		8183,
		item->name);
}
//openssl aes-128-cbc -d -K 56ba77419274d5bd24da592bc415abf0 -in ./file1.ts -out f1.ts -nosalt -iv a9b8013f809f34c03089920ff34aef7e
//openssl enc -aes-128-ecb -k "1122334455667788" -in server.sh -out server.sh.aes
static UM_VOID aecFile(Trancode* ptnc, S_TS_BLOCK* item)
{
	UM_CHAR cmd[1024];
	umapi_sprintf(cmd, "openssl enc -aes-128-cbc -K %02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x"
		" -in /tmp/ts/%s -out /tmp/ts/%s.ts -iv %02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x",
		ptnc->key[0], ptnc->key[1],ptnc->key[2],ptnc->key[3],
		ptnc->key[4], ptnc->key[5],ptnc->key[6],ptnc->key[7],
		ptnc->key[8], ptnc->key[9],ptnc->key[10],ptnc->key[11],
		ptnc->key[12], ptnc->key[13],ptnc->key[14],ptnc->key[15],
		item->name, item->name,
		ptnc->iv[0], ptnc->iv[1],ptnc->iv[2],ptnc->iv[3],
		ptnc->iv[4], ptnc->iv[5],ptnc->iv[6],ptnc->iv[7],
		ptnc->iv[8], ptnc->iv[9],ptnc->iv[10],ptnc->iv[11],
		ptnc->iv[12], ptnc->iv[13],ptnc->iv[14],ptnc->iv[15]);
	runcmd(cmd);
}
#define MAX_TS_BLOCK_COUNT 6
#define TS_DURATION 500
static UM_VOID updateBlocks(Trancode* ptnc)
{
	if(ptnc->curBlock->tmpFile)
		umapi_file_fclose(ptnc->curBlock->tmpFile);
	ptnc->curBlock->tmpFile = UM_NULL;
	if(ptnc->enckey)
		aecFile(ptnc, ptnc->curBlock);
	UM_CHAR m3u8Buff[4096] = { 0 };
	UM_CHAR m3u8KeyLine[400] = { 0 };
	if(ptnc->enckey)
	{
#if USEREMOTEKEYSERVER	
		umapi_sprintf(m3u8KeyLine, 
			"#EXT-X-KEY:METHOD=AES-%d,URI=\"http://%s:%d%s%s\""
			",IV=0x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x\r\n",
			/*ptnc->keytype, */
			128,
			gSystemData.hlsHost,
			gSystemData.hlsPort,
			gSystemData.hlshandlePath,
			getPlayId(ptnc->dest),
			ptnc->iv[0], ptnc->iv[1],ptnc->iv[2],ptnc->iv[3],
			ptnc->iv[4], ptnc->iv[5],ptnc->iv[6],ptnc->iv[7],
			ptnc->iv[8], ptnc->iv[9],ptnc->iv[10],ptnc->iv[11],
			ptnc->iv[12], ptnc->iv[13],ptnc->iv[14],ptnc->iv[15]);
#else
		umapi_sprintf(m3u8KeyLine, 
			"#EXT-X-KEY:METHOD=AES-%d,URI=\"http://%s:%d%s%s\""
			",IV=0x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x\r\n",
			/*ptnc->keytype, */
			128,
			gSystemData.appPulicHost,
			8183,
			ptnc->dest,
			"/k",
			ptnc->iv[0], ptnc->iv[1],ptnc->iv[2],ptnc->iv[3],
			ptnc->iv[4], ptnc->iv[5],ptnc->iv[6],ptnc->iv[7],
			ptnc->iv[8], ptnc->iv[9],ptnc->iv[10],ptnc->iv[11],
			ptnc->iv[12], ptnc->iv[13],ptnc->iv[14],ptnc->iv[15]);		
#endif
	}
	else
		m3u8KeyLine[0] = '\0';
	
	umapi_sprintf(m3u8Buff, "#EXTM3U\r\n"
			"#EXT-X-MEDIA-SEQUENCE:%d\r\n"
			"%s"				
			"#EXT-X-TARGETDURATION:1\r\n",
			ptnc->sequence++,
			m3u8KeyLine
			);
	UM_S32 i;
	for(i = 3; i < MAX_TS_BLOCK_COUNT ; i++)
	{
		updateM3u8(ptnc->block[(ptnc->blockIndex + i) % MAX_TS_BLOCK_COUNT], m3u8Buff);
	}
	S_HLSSERVER * s = ptnc->s;
	umapi_os_mutex_lock(s->mutexHandle, -1);
	struct tm *tm;
	time_t ti;
	char buf2[40];
	ti = time(NULL);
	tm = gmtime(&ti);
	buf2[0] = '\0';
	strftime(buf2, sizeof(buf2), "%a, %d %b %Y %H:%M:%S GMT", tm);
	umapi_snprintf(ptnc->m3u8, sizeof(ptnc->m3u8), 
		"HTTP/1.1 200 OK\r\n"
		"Server: %s\r\n"
		"Date: %s\r\n"
		"Cache-Control: no-cache\r\n"
		"Content-Length: %d\r\n\r\n%s",
		HLSSERVERNAME, buf2, umapi_strlen(m3u8Buff), m3u8Buff);
	writeM3u8File(m3u8Buff, ptnc->dest);
	umapi_os_mutex_unlock(s->mutexHandle);	
}
static UM_S32 writeClient(UM_VOID *s, UM_VOID* n,  AVPacket *pkt)
{
	Trancode* ptnc = (Trancode*)n;
	if(ptnc->curBlock == UM_NULL)
	{
		ptnc->curBlock = ptnc->block[ptnc->blockIndex++];
		if(ptnc->blockIndex >= MAX_TS_BLOCK_COUNT)
			ptnc->blockIndex = 0;
		initBlock(ptnc->curBlock, ptnc);
	}
	if(ptnc->curBlock->tmpFile)
		umapi_file_fwrite(pkt->data, pkt->size, 1, ptnc->curBlock->tmpFile);
	ptnc->curBlock->len += pkt->size;
	ptnc->curBlock->duration = umapi_os_get_tick() - ptnc->curBlock->tick ;
	
	umapi_os_mutex_lock(ptnc->mutexHandle,-1);
	struct umapi_list_head *pos, *next;
	umapi_list_for_each_safe(pos, next, &ptnc->ch)
	{
		S_HLSNODE *tc = umapi_list_entry(pos,S_HLSNODE, head);
		tc->efd->lasttick = umapi_os_get_tick();	
	}
	umapi_os_mutex_unlock(ptnc->mutexHandle);
	if(ptnc->curBlock->duration >= TS_DURATION)
	{
		updateBlocks(ptnc);		
		ptnc->curBlock = UM_NULL;
	}
	return UM_SUCCESS;
}
static UM_S32 readClient(UM_HANDLE fd, UM_VOID *param);
static UM_S32 disConnectClient(UM_HANDLE fd, UM_VOID *param);

static UM_S32 connectClient(UM_HANDLE fd, UM_VOID *param
		,UM_EFD** pefd
	)
{
	LOGI("new connect fd:%d", fd);
	if(fd < 0)
		return UM_FAILURE;
	S_HLSNODE *tc = (S_HLSNODE *)umapi_malloc(sizeof(S_HLSNODE));
	if(!tc)
	{
		LOGS("malloc http sink node fail");
		return UM_FAILURE;
	}
	umapi_memset(tc,0,sizeof(S_HLSNODE));
	tc->s =  ((UM_EFD*)param)->pri;
	tc->fd = fd;
	tc->transcode = UM_NULL;
	UM_EFD *p = um_ufd_malloc(fd, UM_NULL, readClient, disConnectClient, 0x40 | 1, tc);
	if(!p)
	{
		umapi_free(tc);
		LOGS("malloc ufd fail");
		return UM_FAILURE;
	}
	tc->efd = p;
	*pefd = p;
	return UM_SUCCESS;

}
static UM_VOID sendM3u8(UM_HANDLE fd, Trancode * pNode)
{
	LOGI("%s", pNode->m3u8);
	sendReply(fd, "%s", pNode->m3u8);
}
static UM_VOID sendKey(UM_HANDLE fd, Trancode* pNode)
{
	struct tm *tm;
	time_t ti;
	char buf2[40];
	ti = time(NULL);
	tm = gmtime(&ti);
	buf2[0] = '\0';
	strftime(buf2, sizeof(buf2), "%a, %d %b %Y %H:%M:%S GMT", tm);
	if(pNode->keytype != 128)
	{
		LOGE("err keyType");
		pNode->keytype = 128;
	}
	UM_CHAR keyReplybuff[200];
	umapi_sprintf(keyReplybuff, "HTTP/1.1 200 OK\r\n"
		"Server: %s\r\n"
		"Date: %s\r\n"
		"Cache-Control: no-cache\r\n"
		"Content-Length: %d\r\n\r\n",
		HLSSERVERNAME, buf2,  
		pNode->keytype/8);
	UM_S32 len = umapi_strlen(keyReplybuff);
	umapi_memcpy(&keyReplybuff[len], pNode->key, pNode->keytype/8);
	int ret = umapi_socket_send(fd, keyReplybuff, len + pNode->keytype/8, 0);
	if(ret <= 0)
		LOGE("send key ret:%x errno:%d", ret, errno);
	else
	{
		LOGE("send key fd:%d ret:%d keyReplybuff:%s", fd, ret, keyReplybuff);
	}
}
static Trancode *findTsKeyTrancode(S_HLSSERVER *s, S_HLSNODE *tc, UM_CHAR *url)
{
	S_HLSNODE* rt = UM_NULL;
	S_HLSSERVER * c = s;
	umapi_os_mutex_lock(c->mutexHandle, -1);
	Trancode* t;
	UM_CHAR filename[256];
	UM_CHAR *p = umapi_strstr(url,"?");
	if(p)
	{
		umapi_sprintf(filename, "%s", url);
		filename[p - url - 2] = '\0';
	}
	else
	{
		umapi_sprintf(filename, "%s", url);
		filename[umapi_strlen(url) - 2] = '\0';
	}
	hashmap_get(c->map, filename, &t);
	umapi_os_mutex_unlock(c->mutexHandle);
	if(t)
	{
		return t;
	}
	else
		return UM_NULL;
}
static UM_VOID addTsToTrancode(S_HLSNODE * pNode, UM_EFD *efd)
{
	
}
static S_HLSNODE* addToTrancode(S_HLSSERVER *s, S_HLSNODE *tc, UM_CHAR *url)
{
	S_HLSNODE* rt = UM_NULL;
	S_HLSSERVER * c = s;
	umapi_os_mutex_lock(c->mutexHandle, -1);
#if USE_HASH
	Trancode* t;
	UM_CHAR filename[256];
	UM_CHAR *p = umapi_strstr(url,"?");
	if(p)
	{
		umapi_sprintf(filename, "%s", url);
		filename[p - url] = '\0';
	}
	else
	{
		umapi_sprintf(filename, "%s", url);
	}
	hashmap_get(c->map, filename, &t);
	if(t)
	{
		if(tc->transcode != t)
		{
			tc->s = s;
			tc->transcode = t;
			LOGI("add to Trancode fd:%d to url:%s, mutexHandle:%p", 
				tc->fd, 
				((Trancode*)tc->transcode)->dest, t->mutexHandle);
			umapi_os_mutex_lock(t->mutexHandle,-1);
			umapi_list_add(&tc->head, &t->ch);
			umapi_os_mutex_unlock(t->mutexHandle);
		}
		rt = tc;
	}
/*	else
	{
		LOGI("no find :%s", url);
		sendNoFound(tc->fd, "umHlsServer");
	}
*/	
#else
	struct umapi_list_head *pos, *n;
	umapi_list_for_each_safe(pos, n, &c->paths)
	{
	
		Trancode* t = umapi_list_entry(pos,Trancode, head);
		LOGI("t:%p", t);
		if(umapi_strncmp(url, t->dest, umapi_strlen(t->dest)) == 0
			/*
			|| umapi_strncmp(&url[1], t->dest, umapi_strlen(t->dest)) == 0
			*/)
		{	
			tc->server = s;
			tc->transcode = t;
			LOGI("add to Trancode fd:%d to url:%s, mutexHandle:%p", 
				tc->fd, 
				((Trancode*)tc->transcode)->dest, t->mutexHandle);
			umapi_os_mutex_lock(t->mutexHandle,-1);
			umapi_list_add(&tc->head, &t->ch);
			umapi_os_mutex_unlock(t->mutexHandle);
			rt = tc;
			break;
		}
		else
		{
			LOGI("nomatch :%s %s", t->dest, url);
		}
	}
#endif	
	umapi_os_mutex_unlock(c->mutexHandle);
	return rt;
}

static UM_S32 readClient(UM_HANDLE fd, UM_VOID *param)
{
	UM_EFD *efd = (UM_EFD *)param;
	if(!efd || !efd->active)
	{
		LOGE("efd active is false");
		return UM_FAILURE;
	}
	S_HLSSERVER *s = ((S_HLSNODE*)efd->pri)->s;
	HttpHeaders headers = {0};
	headers.contentLen = 0;
	headers.body = UM_NULL;
	if(readHttpPack(fd, &headers) != UM_SUCCESS)
	{
		LOGE("read http Head Fail :%d errno:%d", fd, errno);
		((UM_EFD *)param)->status = 0;
		return UM_FAILURE;
	}
	if(headers.contentLen >0)
	{
		headers.body = (UM_U8 *)umapi_malloc(headers.contentLen + 1);
		if(!headers.body)
		{
			LOGS("malloc body fail");
		}
		else
			umapi_socket_recv(fd, headers.body, headers.contentLen, 0);
	}
	S_HLSNODE* tc = addToTrancode(s, ((UM_EFD *)param)->pri, headers.url);
	if(tc
		&& tc->transcode		
	)
	{
		sendM3u8(fd, tc->transcode);
	}
	else 
	{
		Trancode *t = findTsKeyTrancode(s, ((UM_EFD *)param)->pri, headers.url);
		if(t)
		{
			LOGI("send m3u8 key file %s ", headers.url);
			sendKey(fd, t);	
		}
		else
			sendNoFound(fd, "umHlsServer");
	}
	freeHttpHeaders(&headers);
	return UM_SUCCESS;
}
static UM_S32 disConnectClient(UM_HANDLE fd, UM_VOID *param)
{
	S_HLSNODE *tc  =  ((UM_EFD *)param)->pri;
	Trancode* t = tc->transcode;
	if (t)
	{
		LOGI("begin disconnect %d from Trancode{ %p, %s, %p}", fd, t, t->dest,  t->mutexHandle);
		umapi_os_mutex_lock(t->mutexHandle, -1);
		umapi_list_del(&tc->head);
		umapi_os_mutex_unlock(t->mutexHandle);
		LOGI("end disconnect %d from Trancode{ %p, %s, %p}", fd, t, t->dest,  t->mutexHandle);
	}
	umapi_free(tc);
}


static UM_VOID serverTask(UM_VOID *param)
{
	S_HLSSERVER *s = (S_HLSSERVER*)param;
	UMAPI_SOCKET_ADDR ipaddr;
	ipaddr.addr.ipv4 = 0;
	ipaddr.sin_port = s->port == 8183 ? 29999 : s->port;
	LOGI("start http Listen %d", ipaddr.sin_port);
	startEpoll(s->epoll, &ipaddr, 
		connectClient, UM_NULL, disConnectClient, 
		s,&s->fd
	);
}

static S_HLSSERVER * findHlsServer(S_HLSSINK *s)
{
	initServers();
	if(firstTsBuff == UM_NULL)
	{
		initFirstTsBuff();
	}
	S_HLSSERVER *node = UM_NULL;
	umapi_os_mutex_lock(serverMutexHandle,-1);
	struct umapi_list_head *pos, *n;
	umapi_list_for_each_safe(pos,n, &servers)
	{
		node = umapi_list_entry(pos,S_HLSSERVER,head);
		if(node->port == s->port)
		{
			break;
		}
		node = UM_NULL;
	}
	umapi_os_mutex_unlock(serverMutexHandle);
	if(!node)
	{
		node = (S_HLSSERVER *)umapi_malloc(sizeof(S_HLSSERVER));
		if(!node)
		{
			LOGS("malloc hls sink fail");
			return UM_NULL;
		}
		node->port = s->port;
		umapi_sprintf(node->host, "%s" ,s->host);
		umapi_sprintf(node->path, "%s" ,s->path);
		node->workCall.add = addClient;
		node->workCall.del = delClient;
		node->workCall.stop = stopClient;
		node->workCall.start = startClient;
		node->workCall.write = writeClient;
#if USE_HASH
		node->map = hashmap_new();
#endif		
		umapi_os_mutex_lock(serverMutexHandle,-1);
		umapi_list_add(&node->head, &servers);
		umapi_list_create(&node->clients);
		umapi_os_mutex_create(UM_NULL, UM_OS_MUTEX_ATTR_TIMED_NP, &node->mutexHandle);
		serverTask(node);
		umapi_os_mutex_unlock(serverMutexHandle);
		goto CREATESUCCESS;
CREATEFAIL:
		umapi_free(node);
		node = UM_NULL;		
	}
CREATESUCCESS:	
	return node;
}


SINKFUNS(S_HLSSINK, findHlsServer);

	
S_SINK hlsSink = 
{
	.protocol = "hls",
	.open = open,
	.destroy = destroy,
	.start = start,
	.stop = stop,
	.writePacket = writePacket,
	.priSize = sizeof(S_HLSSINK),
};


